package com.alison.datastream.etl;

import com.alison.datastream.domain.TaxiRide;
import com.alison.datastream.util.TaxiRideGenerator;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author alison
 * @Date 2024/4/4 0:01
 * @Version 1.0
 * @Description keyby--> group by
 */
public class ETLAggreApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<TaxiRide> rides = env.addSource(new TaxiRideGenerator());
        //.keyBy(keySelector)
//        rides.keyBy(ride -> ride.rideId).print();
//        env.execute();
        rides.flatMap((TaxiRide ride, Collector<Tuple2<Long, Long>> out) -> {
            if (!ride.isStart) {
                out.collect(new Tuple2<>(ride.rideId, ride.getEventTimeMillis()));
            }
        }, Types.TUPLE(Types.LONG, Types.LONG)).keyBy(value -> value.f0).print();
        env.execute();
    }
}
